package com.bw.yk08;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.yk09.OrderInfoCargoWide;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Test04 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);


        DataStreamSource<String> orderInfoDs=env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_order_info_topic","test03"));
        DataStreamSource<String> orderDetailDs=env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_order_detail_topic","test03"));


        SingleOutputStreamOperator<OrderInfo> orderInfoMap=orderInfoDs.map(new MapFunction<String, OrderInfo>() {
            @Override
            public OrderInfo map(String s) throws Exception {
                String data=JSON.parseObject(s).getJSONObject("data").toJSONString();
                OrderInfo orderInfo=JSON.parseObject(data,OrderInfo.class);
                SimpleDateFormat sf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                Long time=sf.parse(orderInfo.getCreate_time()).getTime();
                orderInfo.setCreate_ts(time);
                return orderInfo;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                    @Override
                    public long extractTimestamp(OrderInfo orderInfo, long l) {
                        return orderInfo.getCreate_ts();
                    }
                })
        );

        SingleOutputStreamOperator<OrderDetail> orderDetailMap=orderInfoDs.map(new MapFunction<String, OrderDetail>() {
            @Override
            public OrderDetail map(String s) throws Exception {
                String data=JSON.parseObject(s).getJSONObject("data").toJSONString();
                OrderDetail orderDetail=JSON.parseObject(data,OrderDetail.class);
                SimpleDateFormat sf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                Long time=sf.parse(orderDetail.getCreate_time()).getTime();
                orderDetail.setCreate_ts(time);
                return orderDetail;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderDetail>() {
                    @Override
                    public long extractTimestamp(OrderDetail orderDetail, long l) {
                        return orderDetail.getCreate_ts();
                    }
                })
        );


        SingleOutputStreamOperator<OrderInfoWide> joinMap=orderInfoMap.keyBy(x ->x.getId())
                        .intervalJoin(orderDetailMap.keyBy(x -> x.getOrder_info_id()))
                        .between(Time.seconds(-5),Time.seconds(5))
                        .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderInfoWide>() {
                                    @Override
                                    public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, ProcessJoinFunction<OrderInfo, OrderDetail, OrderInfoWide>.Context context, Collector<OrderInfoWide> collector) throws Exception {
                                        collector.collect(new OrderInfoWide(orderInfo, orderDetail));;
                                    }
                        });



        SingleOutputStreamOperator<OrderInfoWide> result= AsyncDataStream.unorderedWait(joinMap, new MyAsyncFunction(), 1000, TimeUnit.MILLISECONDS, 100);

        result.map(JSON::toJSONString).addSink(MyKafkaUtil.getFlinkKafkaProducer("dws_trade_orders"));

        env.execute();
    }
}
